package com.dkx.springbootwebsocket.service;
import com.dkx.springbootwebsocket.entity.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.user.SimpUser;
import org.springframework.messaging.simp.user.SimpUserRegistry;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;


import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class WebSocketMessageService {



    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    @Qualifier("webSocketRedisTemplate") // 使用自定义的RedisTemplate
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private SimpUserRegistry simpUserRegistry;

    // 常量定义 - 添加内存保护限制
    private static final String OFFLINE_MESSAGE_PREFIX = "offline:msg:";
    private static final String USER_SESSION_PREFIX = "user:session:";
    private static final String USER_CONNECTION_COUNT_PREFIX = "user:conn:count:";

    // 内存保护限制
    private static final int MAX_OFFLINE_MESSAGES_PER_USER = 500;        // 每个用户最多500条离线消息
    private static final int MAX_BROADCAST_STORAGE_USERS = 1000;         // 广播消息最多存储给1000个用户
    private static final int MAX_GROUP_USERS = 1000;                      // 群发消息最多1000个用户
    private static final long OFFLINE_MESSAGE_TTL_DAYS = 7;              // 离线消息保存7天
    private static final long USER_SESSION_TTL_MINUTES = 120;           // 2小时
    private static final long USER_CONNECTION_TTL_MINUTES = 120;        // 2小时
    private static final long USER_SESSION_TTL_DAYS = 1;                // 用户会话保存1天

    // 内存中的连接计数和监控
    private final Map<String, AtomicInteger> userConnectionCounts = new ConcurrentHashMap<>();
    private final AtomicInteger totalMessagesSent = new AtomicInteger(0);
    private final long startupTime = System.currentTimeMillis();

    /**
     * 发送点对点消息
     */
    public void sendToUser(String fromUser, String toUser, String content) {
        try {
            // 参数校验
            if (fromUser == null || toUser == null || content == null) {
                log.warn("点对点消息参数为空: from={}, to={}", fromUser, toUser);
                return;
            }

            // 内容长度限制
            if (content.length() > 1000) {
                content = content.substring(0, 1000) + "...";
                log.warn("消息内容过长，已截断: from={}, to={}", fromUser, toUser);
            }

            Message message = createMessage(fromUser, toUser, content, Message.MessageType.P2P);
            sendMessage(message);
            totalMessagesSent.incrementAndGet();

        } catch (Exception e) {
            log.error("发送点对点消息失败: from={}, to={}", fromUser, toUser, e);
        }
    }

    /**
     * 发送群组消息  （内存保护）
     */
    public void sendToGroup(String fromUser, List<String> toUsers, String content) {
        if (toUsers == null || toUsers.isEmpty()) {
            log.warn("群组消息接收者为空: from={}", fromUser);
            return;
        }

        // 限制群发用户数量，防止内存溢出
        if (toUsers.size() > MAX_GROUP_USERS) {
            toUsers = toUsers.subList(0, MAX_GROUP_USERS);
            log.warn("群发消息用户数量超过限制({})，已截断前{}个用户: from={}",
                    MAX_GROUP_USERS, MAX_GROUP_USERS, fromUser);
        }

        // 内容长度限制
        if (content.length() > 1000) {
            content = content.substring(0, 1000) + "...";
            log.warn("群组消息内容过长，已截断: from={}", fromUser);
        }

        Message templateMessage = createMessage(fromUser, "group", content, Message.MessageType.GROUP);
        int successCount = 0;

        for (String toUser : toUsers) {
            try {
                Message userMessage = cloneMessageForUser(templateMessage, toUser);
                sendMessage(userMessage);
                successCount++;
            } catch (Exception e) {
                log.error("发送群组消息失败给用户: {}", toUser, e);
            }
        }

        totalMessagesSent.addAndGet(successCount);
        log.info("群组消息发送完成: from={}, 成功={}, 总数={}", fromUser, successCount, toUsers.size());
    }

    /**
     * 发送广播消息  （内存保护）
     */
    public void sendToAll(String fromUser, String content) {
        try {
            // 内容长度限制
            if (content.length() > 1000) {
                content = content.substring(0, 1000) + "...";
                log.warn("广播消息内容过长，已截断: from={}", fromUser);
            }

            Message message = createMessage(fromUser, "all", content, Message.MessageType.BROADCAST);

            // 1. 首先发送给所有在线用户
            messagingTemplate.convertAndSend("/topic/broadcast", message);
            log.info("广播消息实时发送成功: from={}, content={}", fromUser, content);

            // 2. 异步存储给离线用户（防止阻塞主线程）
            asyncStoreBroadcastForOfflineUsers(message);

            totalMessagesSent.incrementAndGet();

        } catch (Exception e) {
            log.error("发送广播消息失败", e);
        }
    }

    /**
     * 异步存储广播消息给离线用户
     */
    @Async
    protected void asyncStoreBroadcastForOfflineUsers(Message message) {
        try {
            List<String> allUsers = getAllUsers();
            int storedCount = 0;

            for (String user : allUsers) {
                // 跳过发送者自己
                if (user.equals(message.getFromUser())) {
                    continue;
                }

                // 限制存储用户数量，防止内存溢出
                if (storedCount >= MAX_BROADCAST_STORAGE_USERS) {
                    log.warn("广播消息存储用户数量达到上限({}): from={}",
                            MAX_BROADCAST_STORAGE_USERS, message.getFromUser());
                    break;
                }

                // 如果用户不在线，存储广播消息
                if (!isUserOnline(user)) {
                    storeOfflineMessageForUser(user, message);
                    storedCount++;
                }
            }

            log.info("广播消息已存储给 {} 个离线用户: from={}", storedCount, message.getFromUser());
        } catch (Exception e) {
            log.error("异步存储广播消息失败", e);
        }
    }

    /**
     * 发送消息  （统一入口）
     */
    private void sendMessage(Message message) {
        try {
            if (Message.MessageType.BROADCAST.equals(message.getType())) {
                // 广播消息直接发送，离线存储已在sendToAll中处理
                return;
            } else {
                boolean isOnline = isUserOnline(message.getToUser());

                if (isOnline) {
                    // 用户在线，直接发送
                    sendOnlineMessage(message);
                } else {
                    // 用户离线，存储消息
                    storeOfflineMessage(message);
                    log.debug("用户离线，消息已存储: from={}, to={}",
                            message.getFromUser(), message.getToUser());
                }
            }
        } catch (Exception e) {
            log.error("消息发送失败: {}", message, e);
            // 发送失败也存储为离线消息
            storeOfflineMessage(message);
        }
    }

    /**
     * 发送在线消息
     */
    private void sendOnlineMessage(Message message) {
        try {
            messagingTemplate.convertAndSendToUser(
                    message.getToUser(),
                    "/queue/messages",
                    message
            );
            log.debug("在线消息发送成功: to={}", message.getToUser());
        } catch (Exception e) {
            log.error("在线消息发送失败: to={}", message.getToUser(), e);
            throw e;
        }
    }

    /**
     * 存储离线消息  （限制数量，内存保护）
     */
    private void storeOfflineMessage(Message message) {
        String key = OFFLINE_MESSAGE_PREFIX + message.getToUser();
        try {
            // 检查当前消息数量，如果超过限制则清理旧消息
            Long currentSize = redisTemplate.opsForList().size(key);
            if (currentSize != null && currentSize >= MAX_OFFLINE_MESSAGES_PER_USER) {
                // 移除最旧的一条消息
                redisTemplate.opsForList().rightPop(key);
                log.warn("用户离线消息达到上限({})，移除最旧消息: userId={}",
                        MAX_OFFLINE_MESSAGES_PER_USER, message.getToUser());
            }

            // 存储新消息
            redisTemplate.opsForList().leftPush(key, message);
            redisTemplate.expire(key, Duration.ofDays(OFFLINE_MESSAGE_TTL_DAYS));

        } catch (Exception e) {
            log.error("存储离线消息失败: key={}", key, e);
        }
    }

    /**
     * 检查用户是否在线
     */
    public boolean isUserOnline(String userId) {
        if (userId == null) return false;

        String key = USER_SESSION_PREFIX + userId;
        try {
            Boolean exists = redisTemplate.hasKey(key);
            return exists != null && exists;
        } catch (Exception e) {
            log.error("检查用户在线状态失败: userId={}", userId, e);
            return false;
        }
    }

    /**
     * 处理用户上线
     */
    public void handleUserOnline(String userId, String sessionId) {
        if (userId == null || sessionId == null) return;

        String key = USER_SESSION_PREFIX + userId;
        try {
            // 更新用户会话
            redisTemplate.opsForValue().set(key, sessionId,
                    Duration.ofMinutes(USER_SESSION_TTL_MINUTES));

//            redisTemplate.opsForValue().set(key, sessionId, Duration.ofMinutes(USER_SESSION_TTL_DAYS));

            // 更新连接计数
            updateConnectionCount(userId, 1);

            log.debug("用户上线: userId={}, sessionId={}", userId, sessionId);
        } catch (Exception e) {
            log.error("处理用户上线失败: userId={}", userId, e);
        }
    }

    /**
     * 处理用户下线
     */
    public void handleUserOffline(String userId) {
        if (userId == null) return;

        String key = USER_SESSION_PREFIX + userId;
        try {

            // 更新连接计数
            int remainingConnections = updateConnectionCount(userId, -1);

            // 如果连接数为0，删除会话
            if (remainingConnections <= 0) {
                redisTemplate.delete(key);
                userConnectionCounts.remove(userId);
                log.debug("用户完全下线: userId={}", userId);
            } else {
                log.debug("用户部分连接断开，剩余连接数: userId={}, count={}", userId, remainingConnections);
            }
        } catch (Exception e) {
            log.error("处理用户下线失败: userId={}", userId, e);
        }
    }

    /**
     * 更新用户连接计数
     */
    private int updateConnectionCount(String userId, int delta) {
        try {
            String countKey = USER_CONNECTION_COUNT_PREFIX + userId;

            // 更新Redis中的计数
            Long newCount = redisTemplate.opsForValue().increment(countKey, delta);
            if (newCount != null && newCount <= 0) {
                redisTemplate.delete(countKey);
                return 0;
            }

            // 同时更新内存中的计数作为备份
            AtomicInteger memoryCount = userConnectionCounts.computeIfAbsent(userId, k -> new AtomicInteger(0));
            int memoryValue = memoryCount.addAndGet(delta);
            if (memoryValue <= 0) {
                userConnectionCounts.remove(userId);
            }

            // 设置过期时间
            if (newCount != null && newCount > 0) {
                // 设置连接计数的过期时间
                redisTemplate.expire(countKey,
                        Duration.ofMinutes(USER_CONNECTION_TTL_MINUTES));

            }

            return newCount != null ? newCount.intValue() : memoryValue;
        } catch (Exception e) {
            log.error("更新用户连接计数失败: userId={}, delta={}", userId, delta, e);
            // 回退到内存计数
            AtomicInteger memoryCount = userConnectionCounts.computeIfAbsent(userId, k -> new AtomicInteger(0));
            return memoryCount.addAndGet(delta);
        }
    }

    /**
     * 延迟发送离线消息
     */
    @Async
    public void delayedSendOfflineMessages(String userId) {
        try {
            int maxRetries = 3; // 减少重试次数
            int retryInterval = 500; // 减少重试间隔

            for (int i = 0; i < maxRetries; i++) {
                Thread.sleep(retryInterval);

                if (isUserSubscribed(userId)) {
                    log.info("用户订阅已建立，开始发送离线消息: userId={}", userId);
                    sendOfflineMessagesWithRetry(userId);
                    return;
                }

                log.debug("等待用户订阅建立，重试 {}/{}: userId={}", i+1, maxRetries, userId);
            }

            log.warn("用户订阅在重试后仍未建立，放弃发送离线消息: userId={}", userId);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("延迟发送离线消息被中断: userId={}", userId);
        } catch (Exception e) {
            log.error("延迟发送离线消息失败: userId={}", userId, e);
        }
    }

    /**
     * 检查用户是否已订阅
     */
    private boolean isUserSubscribed(String userId) {
        try {
            SimpUser user = simpUserRegistry.getUser(userId);
            if (user != null) {
                return user.getSessions().stream()
                        .flatMap(session -> session.getSubscriptions().stream())
                        .anyMatch(sub -> sub.getDestination().contains("/user/" + userId + "/queue/messages"));
            }
            return false;
        } catch (Exception e) {
            log.error("检查用户订阅状态失败: userId={}", userId, e);
            return false;
        }
    }

    /**
     * 发送离线消息 - 带重试机制
     */
    private void sendOfflineMessagesWithRetry(String userId) {
        String key = OFFLINE_MESSAGE_PREFIX + userId;
        try {
            Long size = redisTemplate.opsForList().size(key);
            if (size == null || size == 0) {
                return;
            }

            log.info("开始发送用户 {} 的 {} 条离线消息", userId, size);

            int maxRetries = 2;
            for (int retry = 0; retry < maxRetries; retry++) {
                if (sendOfflineMessagesOnce(userId)) {
                    log.info("用户 {} 的离线消息发送成功", userId);
                    return;
                }

                if (retry < maxRetries - 1) {
                    Thread.sleep(1000 * (retry + 1));
                }
            }

            log.error("用户 {} 的离线消息发送失败，达到最大重试次数", userId);
        } catch (Exception e) {
            log.error("发送离线消息失败: userId={}", userId, e);
        }
    }

    /**
     * 发送离线消息 - 单次尝试（优化性能）
     */
    private boolean sendOfflineMessagesOnce(String userId) {
        String key = OFFLINE_MESSAGE_PREFIX + userId;
        try {
            Long size = redisTemplate.opsForList().size(key);
            if (size == null || size == 0) {
                return true;
            }

            // 限制单次处理的消息数量，防止内存溢出
            int batchSize = Math.min(size.intValue(), 50);
            boolean allSuccess = true;

            for (int i = 0; i < batchSize; i++) {
                Message message = (Message) redisTemplate.opsForList().rightPop(key);
                if (message == null) break;

                try {
                    if (sendOfflineMessage(userId, message)) {
                        log.debug("离线消息发送成功: messageId={}", message.getId());
                    } else {
                        // 发送失败，把消息放回队列
                        redisTemplate.opsForList().rightPush(key, message);
                        allSuccess = false;
                        break;
                    }
                } catch (Exception e) {
                    log.error("发送单条离线消息失败", e);
                    redisTemplate.opsForList().rightPush(key, message);
                    allSuccess = false;
                    break;
                }
            }

            return allSuccess;
        } catch (Exception e) {
            log.error("发送离线消息单次尝试失败: userId={}", userId, e);
            return false;
        }
    }

    /**
     * 发送单条离线消息
     */
    private boolean sendOfflineMessage(String userId, Message message) {
        try {
            messagingTemplate.convertAndSendToUser(userId, "/queue/messages", message);
            return true;
        } catch (Exception e) {
            log.error("发送离线消息失败: userId={}, messageId={}", userId, message.getId(), e);
            return false;
        }
    }

    /**
     * 工具方法
     */
    private Message createMessage(String fromUser, String toUser, String content, Message.MessageType type) {
        return new Message(
                UUID.randomUUID().toString(),
                fromUser,
                toUser,
                content,
                type,
                System.currentTimeMillis(),
                false
        );
    }

    private Message cloneMessageForUser(Message original, String toUser) {
        return new Message(
                UUID.randomUUID().toString(),
                original.getFromUser(),
                toUser,
                original.getContent(),
                original.getType(),
                System.currentTimeMillis(),
                false
        );
    }

    private void storeOfflineMessageForUser(String userId, Message message) {
        try {
            Message userMessage = cloneMessageForUser(message, userId);
            storeOfflineMessage(userMessage);
        } catch (Exception e) {
            log.error("为用户存储离线消息失败: userId={}", userId, e);
        }
    }

    private List<String> getAllUsers() {
        // 从数据库或用户服务获取，这里返回空列表避免内存溢出

//        return list;
        return Collections.emptyList(); //返回空列表避免内存溢出
    }

    /**
     * 监控和统计方法
     */
    public Map<String, Object> getServiceStats() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalMessagesSent", totalMessagesSent.get());
        stats.put("activeConnections", userConnectionCounts.size());
        stats.put("uptime", System.currentTimeMillis() - startupTime);
        stats.put("memoryUsage", Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory());
        return stats;
    }

    /**
     * 清理资源
     */
    public void cleanup() {
        log.info("开始清理WebSocket消息服务资源...");
        userConnectionCounts.clear();
        log.info("WebSocket消息服务资源清理完成");
    }

}